package dataStream;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import dataStream.Person;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * @desc 滚动窗口
 * @Author xuyouchang
 * @Date 2021/10/9
 */
public class TumblingWindow {
    public static void main(String[] args) throws Exception {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20, 12345L));
        data.add( new Person("man", 18, 123712L));
        data.add( new Person("man", 16, 12998790L));
        data.add( new Person("woman", 20, 1827961387L));
        data.add( new Person("woman", 18, 12983789L));
        data.add( new Person("woman", 16, 1287637812L));
        data.add( new Person("woman", 15,897126L));

        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        // 在 watermark 中指定事件时间
        DataStream<Person> dataStream = env.fromCollection(data);
        dataStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Person>forBoundedOutOfOrderness(Duration.ofMillis(20L))
                .withTimestampAssigner((event, timestamp)->event.getTimestamp())
        );

        // 事件时间 window
        dataStream.keyBy(person->person.gender)
                .window(TumblingEventTimeWindows.of(Time.seconds(1000L)))
                .sum("man")
                .print();


        // 处理时间 window
        dataStream.keyBy(person->person.gender).window(TumblingProcessingTimeWindows.of(Time.seconds(1000L))).sum("man").print();

        env.execute();


    }
}
